package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.common.Constant;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @Author lzc
 * @Date 2023/2/8 11:45
 */
public class FlinkSourceUtil {
    public static KafkaSource<String> getKafkaSource(String topic, String groupId){
        return KafkaSource.<String>builder()
            .setBootstrapServers(Constant.KAFKA_BROKERS)
            .setTopics(topic)  // ctrl+shift+u
            .setGroupId(groupId)
            .setProperty("isolation.level", "read_committed") // 读取已提交数据: 当要消费的 kafka 写入的时候, 如果有事务写入, 则消费事务关闭之后的数据
            .setStartingOffsets(OffsetsInitializer.latest()) // 第一次启动的时候, 从这里地方.
            // 当读取 left join 的结果的时候, 里面的 value 会有 null, 这个地方对 null 反序列化的时候会报空指针异常
            //.setValueOnlyDeserializer(new SimpleStringSchema())  // 将来会有 bug, 要修改成自定义反序列化器
            .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                // 把字节数组反序列化
                @Override
                public String deserialize(byte[] message) throws IOException {
                    if (message != null) {
                        return new String(message, StandardCharsets.UTF_8);
                    }
                    return null;  // flink 的数据中不会有 null, 则 null 值会被 flink 的流自动忽略
                }
                // 是否结束流
                // 从 kafka 读数据, 所以永远返回 false
                @Override
                public boolean isEndOfStream(String nextElement) {
                    return false;
                }
                // 返回流中的数据的类型
                @Override
                public TypeInformation<String> getProducedType() {
                    //return Types.STRING();  // 特殊: 只有基本类型, String Tuple 可用
//                    return TypeInformation.of(String.class);  // 比较通用. 只要类型没有泛型都能用
                     // User<A>
                    return TypeInformation.of(new TypeHint<String>(){});  // 最通用: 任何的类型都可以用
                }
            })
            .build();
    }
}
